package com.isyscore.os.metadata.kettle.step;

import com.isyscore.os.core.exception.DataFactoryException;
import com.isyscore.os.core.exception.ErrorCode;
import com.isyscore.os.metadata.enums.KettleDataFlowNodeType;
import com.isyscore.os.metadata.kettle.base.AbstractStep;
import com.isyscore.os.metadata.kettle.base.FlowConfig;
import com.isyscore.os.metadata.kettle.base.Step;
import com.isyscore.os.metadata.kettle.vis.FilterCondition;
import com.isyscore.os.metadata.kettle.vis.FilterRowsNode;
import com.isyscore.os.metadata.kettle.vis.VisNode;
import groovy.lang.Tuple2;
import org.apache.commons.lang.time.DateUtils;
import org.pentaho.di.core.Condition;
import org.pentaho.di.core.Const;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleStepException;
import org.pentaho.di.core.plugins.PluginInterface;
import org.pentaho.di.core.plugins.PluginRegistry;
import org.pentaho.di.core.plugins.StepPluginType;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.core.row.ValueMeta;
import org.pentaho.di.core.row.ValueMetaAndData;
import org.pentaho.di.core.row.ValueMetaInterface;
import org.pentaho.di.core.row.value.ValueMetaBase;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;
import org.pentaho.di.trans.step.errorhandling.StreamInterface;
import org.pentaho.di.trans.steps.filterrows.FilterRowsMeta;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

public class FilterRowsStep extends AbstractStep {

    @Override
    public StepMeta decode(VisNode step, List<DatabaseMeta> databases, TransMeta transMeta, FlowConfig transGraph) throws Exception {
        FilterRowsNode node = (FilterRowsNode)step;
        PluginRegistry registry = PluginRegistry.getInstance();
        PluginInterface sp = registry.findPluginWithId( StepPluginType.class, KettleDataFlowNodeType.FilterRows.name() );
        StepMetaInterface stepMetaInterface = (StepMetaInterface) registry.loadClass( sp );

        FilterRowsMeta filterRowsMeta = (FilterRowsMeta) stepMetaInterface;

        List<StreamInterface> targetStreams = filterRowsMeta.getStepIOMeta().getTargetStreams();

        targetStreams.get(0).setSubject(node.getSendTrueTo());
        targetStreams.get(1).setSubject(node.getSendFalseTo());


        Condition conditions = new Condition();
        conditions.setNegated(false);
        for (FilterCondition nodeCondition : node.getConditions()) {
            Condition condition = new Condition();
            condition.setNegated(false);
            condition.setLeftValuename(nodeCondition.getFiled());
            condition.setFunction(Integer.parseInt(nodeCondition.getFunction().getSymbol()));
            condition.setOperator(nodeCondition.getOperator().getSymbol());

            ValueMetaAndData valueMetaAndData = new ValueMetaAndData();
            ValueMeta valueMeta = new ValueMeta("constant", ValueMetaBase.TYPE_STRING);
            valueMeta.setLength(-1);
            valueMeta.setPrecision(-1);

            valueMetaAndData.setValueData(nodeCondition.getValue());

//            Date dateValue = isDate(nodeCondition.getValue());
//            if(dateValue!=null){
//                valueMetaAndData.setValueData(new SimpleDateFormat("yyyy/MM/dd HH:mm:ss.SSS").format(dateValue));
//            }else{
//                valueMetaAndData.setValueData(nodeCondition.getValue());
//            }
            valueMetaAndData.setValueMeta(valueMeta);
            condition.setRightExact(valueMetaAndData);
            conditions.addCondition(condition);
        }
        filterRowsMeta.setCondition(conditions);
        StepMeta stepMeta = new StepMeta(KettleDataFlowNodeType.FilterRows.name(), node.getNodeName(), stepMetaInterface);
        stepMeta.setDraw(true);

        return stepMeta;
    }


    public StepMeta after(VisNode step, List<DatabaseMeta> databases, TransMeta transMeta, FlowConfig transGraph) throws KettleStepException {
        StepMeta stepMeta = getStep(transMeta,step.getNodeName());
        RowMetaInterface prevStepFields = transMeta.getPrevStepFields(stepMeta);
        FilterRowsMeta filterRowsMeta = (FilterRowsMeta) stepMeta.getStepMetaInterface();

        List<ValueMetaInterface> valueMetaList = prevStepFields.getValueMetaList();
        List<String> fields = valueMetaList.stream().map(v -> v.getName()).collect(Collectors.toList());
        for (Condition condition : filterRowsMeta.getCondition().getChildren()) {
            if(fields.contains(condition.getLeftValuename()) && "Date".equals(getFiledType(valueMetaList,condition.getLeftValuename()))){
                Date dateValue = isDate(String.valueOf(condition.getRightExact().getValueData()));
                if(dateValue!=null){
                    condition.getRightExact().setValueData(new SimpleDateFormat("yyyy/MM/dd HH:mm:ss.SSS").format(dateValue));
                    condition.getRightExact().getValueMeta().setType(ValueMetaBase.TYPE_STRING);
                }
            }
        }

            String[] nextStepNames = transMeta.getNextStepNames(stepMeta);
            if(nextStepNames.length>0){
                filterRowsMeta.setTrueStepname(nextStepNames[0]);//默认选第一个作为输出
            }

        return null;
    }


}
